package com.zfzn.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Auther lolineet
 * @Email tj.zhanghongwei@163.com
 * @Date 2021/8/5 15:28
 * @Description word count demo
 */
public class WordCount {
    public static void main(String args[]) throws Exception {
        //create excute environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //从文件中读取数据
//        String inputPath="E:\\全部项目\\flinkdemo\\src\\main\\resources\\hello.txt";
//        DataStreamSource<String> dataSource = env.socketTextStream("localhost",10000);
        DataStream<String> dataStream = env.socketTextStream("localhost",10000);

        //对数据进行处理  按空格分词展开  转换成(word,1)二元组进行统计
        DataStream resultStream=dataStream.flatMap(new MyStringFlatMapper()).keyBy(0).sum(1);
        DataStream resultStream1=dataStream.flatMap(new WordCount1.MyCharFlatMapper()).keyBy(0).sum(1);
        resultStream.print();
        resultStream1.print();

        env.execute();
    }

    public static class MyStringFlatMapper implements FlatMapFunction<String,Tuple2<String,Integer>> {

        @Override
        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            String[] words=s.split(" ");
            for(String word:words){
                collector.collect(new Tuple2<>(word,1));
            }
        }
    }

}
